package com.yihaodian.indices;

import java.util.Collection;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.apache.log4j.Logger;
import org.apache.solr.core.CoreContainer;
import org.apache.solr.core.SolrConfig;
import org.apache.solr.schema.IndexSchema;

import com.google.inject.Inject;
import com.yihaodian.api.client.HedwigServiceProvider;
import com.yihaodian.api.search.SearchException;
import com.yihaodian.cluster.Node;
import com.yihaodian.cluster.Node.Status;
import com.yihaodian.index.IndexShard;
import com.yihaodian.index.data.IndexDataService;
import com.yihaodian.index.meta.IndexMeta;
import com.yihaodian.index.meta.shard.Shard;
import com.yihaodian.index.meta.shard.ShardId;
import com.yihaodian.index.meta.shard.ShardRouting;
import com.yihaodian.index.solr.SolrConfigs;
import com.yihaodian.index.solr.SolrIndexShard;

public class SolrIndicesService implements IndicesService {


	static Logger logger = Logger.getLogger(SolrIndicesService.class);

	/**
	 * IndexShard hosts in the node.
	 */
	Map<ShardId, IndexShard> indexShardMap;

	/**
	 * Shard which register in the hedwig, can be searched.
	 */
	Map<ShardId, ShardRouting> shardRouting;

	/**
	 * Solr Configs
	 */
	Map<String, SolrConfigs> configMap;
	/**
	 * 
	 */
	IndexDataService indexDataService;

	/**
	 * solr container
	 */
	CoreContainer coreContainer = null;

	/**
	 * local node
	 */
	Node localNode = null;

	@Inject
	public SolrIndicesService(Node node, IndexDataService dataService,
			CoreContainer coreContainer) {
		this.localNode = node;
		this.indexDataService = dataService;
		this.coreContainer = coreContainer;
		this.indexShardMap = new ConcurrentHashMap<ShardId, IndexShard>();
		this.shardRouting = new ConcurrentHashMap<ShardId, ShardRouting>();
		this.configMap = new ConcurrentHashMap<String, SolrConfigs>();
	}

	/**
	 * start a index engine and register to hedwig.
	 * 
	 * @param indexId
	 * @param shardId
	 * @param register
	 * @return
	 */
	public synchronized boolean startIndexShard(IndexMeta meta,
			ShardRouting routing) {
		ShardId shardId = routing.getShardId();
		String version = meta.version(shardId);
		String indexDir = indexDataService.getDataPath(shardId, version);

		SolrConfigs configs = meta.getSolrConfigs();
		if (configs == null || configs.getSolrConfig() == null) {
			configs = SolrConfigs.loadFromClassPath(indexDir,
					shardId.getIndexId());
		}
		
		// save solrconfigs
		String index = shardId.getIndexId();
		if(configMap.get(index) == null) {
			configMap.put(index, configs);
		}

		IndexSchema indexSchema = configs.getIndexSchema();
		SolrConfig solrConfig = configs.getSolrConfig();

		// create
		if (indexSchema == null) {
			logger.info("index schema not set yet " + shardId);
			return false;
		}

		if (version == null) {
			logger.info("version is null " + shardId);
			return false;
		}

		IndexShard indexShard = new SolrIndexShard(localNode, shardId,
				indexDir, indexSchema, solrConfig, coreContainer);
		indexShard.startup();
		logger.info("started indexShard: " + shardId);

		// register
		ShardRouting sr = new ShardRouting(routing);
		sr.setVersion(version);
		shardRouting.put(shardId, sr);

		registerIndexShard(shardId, sr.isEnable());

		indexShardMap.put(shardId, indexShard);
		localNode.setStatus(Status.STARTING_SHARD);
		return true;
	}

	public void registerIndexShard(ShardId shard, boolean enable) {
		try {
			logger.info("register " + localNode + " " + shard);
			int port = localNode.getPort();
			String serviceVersion = HedwigServiceProvider
					.shardServiceVersion(shard.getIndexId(), shard.getShardId());
			HedwigServiceProvider.register(
					HedwigServiceProvider.serviceName_shardSearch,
					serviceVersion, port, enable);

			ShardRouting sr = shardRouting.get(shard);
			sr.setEnable(true);
		} catch (Exception e) {
			logger.error("register shard failed " + shard, e);
		}
	}

	/**
	 * enable/disable
	 */
	public void updateIndexShard(ShardId shard, boolean enable) {
		try {
			logger.info("updateIndexShard " + localNode + " " + shard + " " + enable);
			int port = localNode.getPort();
			String serviceVersion = HedwigServiceProvider
					.shardServiceVersion(shard.getIndexId(), shard.getShardId());
			HedwigServiceProvider.updateServiceStatus(
					HedwigServiceProvider.serviceName_shardSearch,
					serviceVersion, port, enable);

			ShardRouting sr = shardRouting.get(shard);
			sr.setEnable(enable);
		} catch (Exception e) {
			logger.error("unregister shard failed " + shard, e);
		}
	}

	@Override
	public void stopIndexShard(ShardId shard) {
		IndexShard indexShard = indexShardMap.get(shard);
		if (indexShard != null) {
			updateIndexShard(shard, false);
			indexShard.shutdown();
			shardRouting.remove(shard);
			indexShardMap.remove(shard);
		}
	}

	@Override
	public IndexShard getIndexShard(ShardId shard) {
		return indexShardMap.get(shard);
	}

	@Override
	public void enableShard(ShardId shardId) {
		ShardRouting sr = shardRouting.get(shardId);
//		if (!sr.isEnable()) {
			updateIndexShard(shardId, true);
//		}
	}

	@Override
	public void disableShard(ShardId shardId) {
		ShardRouting sr = shardRouting.get(shardId);
//		if (sr.isEnable()) {
			updateIndexShard(shardId, false);
//		}
	}

	@Override
	public Collection<ShardRouting> shardRouting() {
		return shardRouting.values();
	}

	@Override
	public void startup() throws SearchException {
		// TODO Auto-generated method stub

	}

	@Override
	public void shutdown() throws SearchException {
		for (IndexShard shard : indexShardMap.values()) {
			shard.shutdown();
		}

	}

	@Override
	public String getSchemaXml(String indexId) {
		String ret = null;
		SolrConfigs config = configMap.get(indexId);
		if(config != null) {
			ret = config.getSchemaXml();
		}
		return ret;
	}
}
